package cn.tedu.simple.utils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;


/**
 * @Author: Miyako
 * @Date: 2024-09-26-10:40
 * @Description:
 */
public class RabbitMQUtil {
    private ConnectionFactory factory = new ConnectionFactory();
    private Channel cc ;
    private String method;
    private Connection nc;
    private Boolean isSave;

    //初始化连接
    public RabbitMQUtil(String username , String password ){
        factory.setHost("192.168.8.100");
        factory.setPort(5672);
        factory.setUsername(username);
        factory.setPassword(password);
        try {
             nc = factory.newConnection();
             cc = nc.createChannel();
        }catch (Exception e){
            System.out.println("连接出错");
            e.printStackTrace();
        }
    }

    //创建简单队列
    public void createSimpleQueue( Boolean isSave , Boolean isOnly , Boolean autoDel , Map<String, Object> other) throws IOException {
        this.isSave = isSave;
        cc.queueDeclare("simple.queue",isSave,isOnly,autoDel,other);
        method="simple.queue";
    }
    //创建工作队列
    public void createWorkQueue(Boolean isSave , Boolean isOnly , Boolean autoDel , Map<String, Object> other) throws IOException {
        this.isSave=isSave;
        cc.queueDeclare("work.queue",isSave,isOnly,autoDel,other);
        method="work.queue";
    }

    //生产者发送消息
    public void sendMessage(String message  ) throws IOException {
        if (isSave){
            //持久化
            cc.basicPublish("",method,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
        }{
            //不持久化
            cc.basicPublish("",method,null,message.getBytes());
        }


    }
    public void closeConnection() throws IOException {
        nc.close();
    }


    //消费者接收消息 autoAck ——>是否自动确认
    public void acceptMessage(Boolean autoAck) throws IOException {
        DeliverCallback d = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //Delivery可以获取要消费的数据
                System.out.println(consumerTag);
                byte[] a = message.getBody();
                String s = new String(a);
                System.out.println(consumerTag + "收到：" + s);
                if (s.contains(".")) {
                    try {
                        System.out.println(consumerTag + "开始处理消息...");
                        //睡眠10s钟
                        Thread.sleep(10000);
                        System.out.println(consumerTag + "处理消息完毕!");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("-----------");

                cc.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        //取消接收时的回调函数
        CancelCallback c = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {

            }
        };
        //设置单个消费者只处理一条消息，实现消息的合理并发
        cc.basicQos(1);
        //第一个参数：指定队列名称
        //第二个参数：先设置为true,设置为Autoack自动确认消息，只要消息发送给消费者
        //就会认为自己的消息处理完毕了，从RabbitMQ中删除，如果消费者还没来的及消费，就宕机了
        //就会造成消息丢失，所以我们不使用自动ack，将第二个值改为false
        //此处设置为false，还没完成ack的确认，消息消费完毕还需还设置ack的主动提交
        //手动ack的模式下：如果消费者还未发送消息回执的标签，这个消息会暂时保存在RabbitMQ中
        //不会由于消费者宕机而造成数据丢失
        //第三个参数：处理数据的回调函数
        //第四个参数：取消接收时的回调函数
        cc.basicConsume(method, autoAck, d, c);
    }


}
